 
/* 
 * Copyright (C) Ignacio Lucas
 */
 
/*
 *  Librería que encpasula una cola FIFO
 *
 *  Ignacio Lucas <ignacio.lucas@gmail.com>
 *
 */
#include "common.h"
#include "priority_queue.h"
#include "queue.h"

#define priority_queue_is_valid_priority(p) (p==MASHUP_SERVER_PRIORITY_QUEUE_HIGH || p==MASHUP_SERVER_PRIORITY_QUEUE_NORMAL)

#define PRIORITY_QUEUE_OP_MODE_COLA_DIRECTA 0
#define PRIORITY_QUEUE_OP_MODE_PRIORIDAD 1

// uso interno. Tantas colas internas como prioridades existan.
#define PRIORITY_QUEUE_COUNT 2

// cada un millon de extracciones, rebobinamos el contador de elementos extraidos.
#define PRIORITY_QUEUE_REWIND_COUNT 1000000

// estructura que representa la cola de prioridad.
typedef struct s_priority_queue
{
	// una cola para cada prioridad.
	queue_t colas[PRIORITY_QUEUE_COUNT];

	// configuracion inicial de la cola. Valores indicados por el cliente. Normalizados
	// para que la suma de los porcentajes sea 100.
	priority_queue_conf_t init_conf;

	// cantidad de elementos extraidos de cada prioridad.
	unsigned long extracted_elements[PRIORITY_QUEUE_COUNT];

	// modo de operacion de la cola. Puede ser COLA_DIRECTA o PRIORIDAD.
	char op_mode;

	// guarda la referencia a la cola de donde sacar elementos cuando estoy en modo cola directa.
	int cache_queue_direct;

	// guarda la referencia a la cola cuyo valor de prioridad esta insatisfecho.
	int cache_queue_priority;

} priority_queue;

priority_queue_t priority_queue_create()
{
	priority_queue *q;
	int i=0,j=0;
	if ((q = (priority_queue *)malloc(sizeof(priority_queue))) == NULL)
	{
		return NULL;
	}
	memset(q,0,sizeof(priority_queue));

	for (i=0; i<PRIORITY_QUEUE_COUNT;i++)
	{
		q->colas[i]=queue_create();
		if (q->colas[i] == NULL)
		{
			for (j=0;j<i;j++)
			{
				free(q->colas[j]);
			}
			free (q);
			return NULL;
		}
	}
	return (priority_queue_t)q;
}

int priority_queue_init(priority_queue_t p, priority_queue_conf_t conf)
{
	priority_queue *pq = (priority_queue *)p;
	/** Verificamos que la configuracion sea correcta. **/
	if ((conf.porcentaje_despacho_high<=0) || (conf.porcentaje_despacho_normal<=0) ||
			(conf.porcentaje_despacho_high < conf.porcentaje_despacho_normal) ||
			((conf.porcentaje_despacho_high + conf.porcentaje_despacho_normal) != 100))
	{
		return 1;
	}
	pq->init_conf.porcentaje_despacho_high=conf.porcentaje_despacho_high;
	pq->init_conf.porcentaje_despacho_normal=conf.porcentaje_despacho_normal;

	// inicialmente el modo de operacion es cola directa (hasta que se vayan cargando las colas).
	pq->op_mode = PRIORITY_QUEUE_OP_MODE_COLA_DIRECTA;
	pq->cache_queue_direct=pq->cache_queue_priority=-1;
	pq->extracted_elements[MASHUP_SERVER_PRIORITY_QUEUE_HIGH]=pq->extracted_elements[MASHUP_SERVER_PRIORITY_QUEUE_NORMAL]=0;
	return 0;
}

/** La invocacion a esta funcion siempre se hace con elementos en ambas prioridades. **/
static inline int priority_queue_get_unsatisfied_priority(priority_queue *pq)
{
	double denominador = (double)(pq->extracted_elements[MASHUP_SERVER_PRIORITY_QUEUE_HIGH]+pq->extracted_elements[MASHUP_SERVER_PRIORITY_QUEUE_NORMAL]);
	double numerador = (double)(100*pq->extracted_elements[MASHUP_SERVER_PRIORITY_QUEUE_HIGH]);

	if (((int)(numerador / denominador)) <= pq->init_conf.porcentaje_despacho_high)
	{
		return MASHUP_SERVER_PRIORITY_QUEUE_HIGH;
	}
	return MASHUP_SERVER_PRIORITY_QUEUE_NORMAL;
}

int priority_queue_push(priority_queue_t q, void *data, int priority)
{
	priority_queue *pq = (priority_queue *)q;
	int ret = 0;
	if (!priority_queue_is_valid_priority(priority))
	{
		return -1;
	}

	/**
	 * El push de un elemento puede cambiar de modo de operación
	 * de cola directa a prioridad si la cola directa en cache no es igual
	 * a la prioridad del nuevo elemento.
	 * A su vez, si mi cache de cola directa es invalido (-1), la prioridad
	 * del elemento entrante me define el caché de cola directa (primer elemento entrante).
	 */
	ret = queue_push(pq->colas[priority],data);
	//printf("\nPUSH COLA %d",priority);
	if ((ret == 0) && (pq->op_mode == PRIORITY_QUEUE_OP_MODE_COLA_DIRECTA))
	{
		if (pq->cache_queue_direct == -1)
		{
			//printf("\nACTIVO CACHE DIRECTO: %d",priority);
			pq->cache_queue_direct = priority;
		}
		else if (pq->cache_queue_direct != priority)
		{
			pq->op_mode = PRIORITY_QUEUE_OP_MODE_PRIORIDAD;
			/** Aqui tengo que precargar la cache de la cola de prioridad que corresponda. **/
			pq->cache_queue_priority = priority_queue_get_unsatisfied_priority(pq);

			//printf("\nACTIVO CACHE PRIORITY: %d",pq->cache_queue_priority);
		}
	}
	return ret;
}

int priority_queue_push_vector(priority_queue_t q, void **data, int size, int priority)
{
	int ret_pushed=0,i=0;
	if (!priority_queue_is_valid_priority(priority))
	{
		return -1;
	}
	for (i=0; i<size; i++)
	{
		if (priority_queue_push(q,data[i],priority) == 0)
		{
			ret_pushed++;
		}
		else
		{
			// error, no se pudo insertar el elemento de la posicion i.
			break;
		}
	}
	return ret_pushed;
}

static inline int priority_queue_get_opposite_priority(int priority)
{
	return (priority + 1)%2;
}

static inline void priority_queue_update_after_extract(priority_queue *pq)
{
	pq->extracted_elements[pq->cache_queue_priority]++;

	/**
	 * Pueden suceder 2 cosas:
	 * - la cola de la cual extraje quedo vacia, por lo que paso al modo cola directa y caché de cola directa en la otra cola
	 *   que sí o sí tiene elementos (sino no podría estar en modo prioridad).
	 * - la cola de donde extraje no se vació pero pudo haberse alterado el porcentaje de extraccion de la cola.
	 */
	if (queue_is_empty(pq->colas[pq->cache_queue_priority]))
	{
		pq->op_mode = PRIORITY_QUEUE_OP_MODE_COLA_DIRECTA;
		pq->cache_queue_direct = priority_queue_get_opposite_priority(pq->cache_queue_priority);
	}
	else
	{
		// recalculo el caché de prioridad.
		pq->cache_queue_priority = priority_queue_get_unsatisfied_priority(pq);
	}

	// normalizo mediciones en caso de haber llegado al rewind. Lo llevamos a que la suma sea aprox. 100.
	if (pq->extracted_elements[MASHUP_SERVER_PRIORITY_QUEUE_HIGH] + pq->extracted_elements[MASHUP_SERVER_PRIORITY_QUEUE_NORMAL] == PRIORITY_QUEUE_REWIND_COUNT)
	{
		pq->extracted_elements[MASHUP_SERVER_PRIORITY_QUEUE_HIGH]=(unsigned long)(pq->extracted_elements[MASHUP_SERVER_PRIORITY_QUEUE_HIGH]*100/PRIORITY_QUEUE_REWIND_COUNT);
		pq->extracted_elements[MASHUP_SERVER_PRIORITY_QUEUE_NORMAL]=(unsigned long)(pq->extracted_elements[MASHUP_SERVER_PRIORITY_QUEUE_NORMAL]*100/PRIORITY_QUEUE_REWIND_COUNT);
	}
}

static inline void *priority_queue_pull_element(priority_queue *pq,int queue)
{
	void *ret = queue_pull(pq->colas[queue]);
	pq->extracted_elements[queue]++;
	return ret;
}

// implementacion del algoritmo de prioridad.
void* priority_queue_pull(priority_queue_t qt)
{
	priority_queue *pq = (priority_queue *)qt;
	void *ret = NULL;

	/** Devolvemos rapido un elemento si no estoy operando en modo prioridad. **/
	if (pq->op_mode==PRIORITY_QUEUE_OP_MODE_COLA_DIRECTA)
	{
		/* Modo cola directa con cache -1 indica que no hay elementos en ninguna cola. **/
		if (pq->cache_queue_direct!=-1)
		{
			//printf("\nSACO DE COLA DIRECTA %d",pq->cache_queue_direct);
			ret = priority_queue_pull_element(pq, pq->cache_queue_direct);
			if (queue_is_empty(pq->colas[pq->cache_queue_direct]))
			{
				pq->cache_queue_direct=-1;
			}
		}
	}
	else
	{
		//printf("\nSACO DE COLA PRIORIDAD %d",pq->cache_queue_priority);
		/** Estoy en modo prioridad. Extraemos de la cola indicada en cache (siempre este caché tendrá valor válido). **/
		ret = priority_queue_pull_element(pq,pq->cache_queue_priority);

		/**
		 * Despues de quitar un elemento en modo prioridad, podria suceder que pase al esquema
		 * de cola directa (se vacio la cola de donde saque el elemento)
		 */
		priority_queue_update_after_extract(pq);
	}
	return ret;
}

// la cola de prioridad estara vacia si no hay elementos en ninguna prioridad.
int priority_queue_is_empty(priority_queue_t q)
{
	priority_queue * theq=(priority_queue *)q;
	return (queue_is_empty(theq->colas[MASHUP_SERVER_PRIORITY_QUEUE_HIGH]))
			&&
			(queue_is_empty(theq->colas[MASHUP_SERVER_PRIORITY_QUEUE_NORMAL]));
}

int priority_queue_destroy(priority_queue_t q)
{
	priority_queue * theq=(priority_queue *)q;

	if (theq != NULL)
	{
		if (theq->colas[MASHUP_SERVER_PRIORITY_QUEUE_HIGH] != NULL)
		{
			queue_destroy(theq->colas[MASHUP_SERVER_PRIORITY_QUEUE_HIGH]);
		}
		if (theq->colas[MASHUP_SERVER_PRIORITY_QUEUE_NORMAL] != NULL)
		{
			queue_destroy(theq->colas[MASHUP_SERVER_PRIORITY_QUEUE_NORMAL]);
		}
		free(theq);
	}
	return 0;
}

int priority_queue_is_valid(int priority)
{
	return (priority == MASHUP_SERVER_PRIORITY_QUEUE_NORMAL) || ((priority == MASHUP_SERVER_PRIORITY_QUEUE_HIGH))?1:0;
}

